
package com.lee.job; 
 
import com.google.common.collect.Lists; 
import com.lee.enumeration.FinishQualifierEnum; 
import com.lee.info.CkFlowInfo; 
import com.lee.info.LkFlowInfo; 
import com.lee.logger.Logger; 
import com.lee.utils.*; 
import org.apache.flink.api.common.functions.FlatMapFunction; 
import org.apache.flink.util.Collector; 
 
import java.time.Duration; 
import java.time.LocalDateTime; 
import java.util.ArrayList; 
import java.util.List; 
import java.util.Objects; 
 
public class ScanDFlowFunction implements FlatMapFunction<CkFlowInfo, LkFlowInfo> { 
 
 
 
    @Override
    public void flatMap(CkFlowInfo ckFlowInfo, Collector<LkFlowInfo> collector) {
        try {
            if(DMapUtils.getIndex() > ConstantsUtils.DATA_BLOCK_SIZE){
                Logger.debug(ckFlowInfo.getDIndex() ,"/",ckFlowInfo.getDCIndex(),"ScanDFlowFunction -> 正在计算汇总结果，退出当前模块  ");
                return;
            }
            Logger.debug(ckFlowInfo.getDIndex() ,"/",ckFlowInfo.getDCIndex(),"ScanDFlowFunction -> ckFlowInfo ", ckFlowInfo);
            LocalDateTime start = LocalDateTime.now();
            List<String> Lk = new ArrayList<>();
            long index = ckFlowInfo.getDIndex();
            long cindex = ckFlowInfo.getDCIndex();
            long num = DCountUtils.getIndex();
            List<List<Long>> sameIndex;
            List<String> CkFlowItem = ckFlowInfo.getCkList();
            if(Objects.isNull(CkFlowItem)){
                return;
            }
            //1、查询CkFlowItem的元素在原数据集中出现的次数
            sameIndex = DMapUtils.getValueByKeyGinix(Lists.newArrayList(CkFlowItem.get(0)));
 
            for (int i = 0; i < CkFlowItem.size(); i++) {
                List<List<Long>> nextItem = DMapUtils.getValueByKeyGinix(Lists.newArrayList(CkFlowItem.get(i)));
                sameIndex = ArrayUtils.getSameItemGinix(sameIndex, nextItem);
            }
 
            //2、从汇总表中查询支持度计数
            long preCount = DCountUtils.getValueByKey(CkFlowItem);
            long count = ArrayUtils.getItemSizeGinix(sameIndex) + preCount;
            //3、根据查询出来的次数更新支持度
            double support = 1.0 * count / num;

 
            SupportDataUtils.setValueByKey(CkFlowItem, support);
            SupportDataUtils.setValueByKeyNum(CkFlowItem, count);
            Logger.debug(index, " check BdsScanDFlowFunction -> CkFlowItem ", CkFlowItem, " len ", count, " num ", num, " support ", support);
 
            if (support >= ConstantsUtils.MIN_SUPPORT && num > ConstantsUtils.MIN_WINDOW_SIZE) {
 
                Logger.info(index, " changed BdsScanDFlowFunction -> CkFlowItem ", CkFlowItem, " len ", count, " num ", num, " support ", support);
                Lk = CkFlowItem;
 
            }
 
 
            LkFlowInfo lkFlowInfo = new LkFlowInfo();
            lkFlowInfo.setDIndex(index);
            lkFlowInfo.setDCIndex(cindex);
            lkFlowInfo.setLkList(Lk);
            lkFlowInfo.setDataFlow(ckFlowInfo.getDataFlow());
            collector.collect(lkFlowInfo);
 
            DMapUtils.addFinishIndex(FinishQualifierEnum.SCAN_D.getCode(),1l);
 
            LocalDateTime end = LocalDateTime.now();
            Duration between = Duration.between(start, end);
            Logger.debug(ckFlowInfo.getDIndex() ,"/",ckFlowInfo.getDCIndex(),between.toMillis(),"END ScanDFlowFunction -> ckFlowInfo ", ckFlowInfo);
 
        }catch(Exception e){
            Logger.error("BdsScanDFlowFunction -> 支持度计算报错：",e);
            e.printStackTrace();
        }
 
    }
} 
